package com.danan.realtime.app.dwd;

import com.danan.realtime.app.BaseSqlApp;
import com.danan.realtime.common.Constant;
import com.danan.realtime.util.SinkUtil;
import com.danan.realtime.util.SourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/1/25
 */
public class DwdApp_5_trade_order_pay_success_detail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdApp_5_trade_order_pay_success_detail().init(4005,3,"DwdApp_5_trade_order_pay_success_detail");
    }
    @Override
    public void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 读取ods_db数据
        readOdsDb(tableEnv,"DwdApp_5_trade_order_pay_success_detail");
        // 2 过滤支付成功的订单数据
        filterPaySucOrderInfo(tableEnv);
        // 3 读取下单明细表
        readOrderDetail(tableEnv);
        // 4 读取字典表
        readBaseDic(tableEnv);
        // 5 join
        join(tableEnv);
        // 6 写出到Kafka中
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        // 创建目标表，与kafka关联
        tableEnv.executeSql("create table target (" +
                "order_detail_id string," +
                "order_id string," +
                "user_id string," +
                "sku_id string," +
                "sku_name string," +
                "province_id string," +
                "activity_id string," +
                "activity_rule_id string," +
                "coupon_id string," +
                "payment_type_code string," +
                "payment_type_name string," +
                "callback_time string," +
                "source_id string," +
                "source_type_code string," +
                "source_type_name string," +
                "sku_num string," +
                "split_original_amount string," +
                "split_activity_amount string," +
                "split_coupon_amount string," +
                "split_payment_amount string," +
                "ts bigint " +
                ")" +
                SinkUtil.getKafkaWith(Constant.TOPIC_DWD_TRADE_PAY_DETAIL_SUC));
        // 数据导入
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void join(StreamTableEnvironment tableEnv) {
        // 设置数据的TTL
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));
        // join
        Table joinResult = tableEnv.sqlQuery("select " +
                "od.id order_detail_id," +
                "od.order_id," +
                "od.user_id," +
                "od.sku_id," +
                "od.sku_name," +
                "od.province_id," +
                "od.activity_id," +
                "od.activity_rule_id," +
                "od.coupon_id," +
                "ps.payment_type payment_type_code," +
                "dic.dic_name payment_type_name," +
                "ps.callback_time," +
                "od.source_id," +
                "od.source_type source_type_code," +
                "od.source_type_name," +
                "od.sku_num," +
                "od.split_original_amount," +
                "od.split_activity_amount," +
                "od.split_coupon_amount," +
                "od.split_total_amount split_payment_amount," +
                "ps.ts " +
                "from payment_info ps " +
                "join order_detail od on ps.order_id = od.order_id " +
                "join base_dic for system_time as of ps.pt as dic on ps.payment_type = dic.dic_code");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void readOrderDetail(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table order_detail (" +
                "id string," +
                "order_id string," +
                "user_id string," +
                "sku_id string," +
                "sku_name string," +
                "province_id string," +
                "activity_id string," +
                "activity_rule_id string," +
                "coupon_id string," +
                "date_id string," +
                "create_time string," +
                "source_id string," +
                "source_type string," +
                "source_type_name string," +
                "sku_num string," +
                "split_original_amount string," +
                "split_activity_amount string," +
                "split_coupon_amount string," +
                "split_total_amount string," +
                "ts bigint " +
                ")" +
                SourceUtil.getKafkaWith(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,"DwdApp_5_trade_order_pay_success_detail"));
    }

    private void filterPaySucOrderInfo(StreamTableEnvironment tableEnv) {
        Table paymentInfo = tableEnv.sqlQuery("select " +
                "data['user_id'] user_id," +
                "data['order_id'] order_id," +
                "data['payment_type'] payment_type," +
                "data['callback_time'] callback_time," +
                "ts," +
                "pt " +
                "from source " +
                "where `database`='gmall2023' and `table`='payment_info' " +
                "and `type`='update' " +
                "and `old`['payment_status'] is not null " +
                "and `data`['payment_status']='1602'");
        tableEnv.createTemporaryView("payment_info",paymentInfo);
    }
}
